package org.zjt.chapter1.http;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.local.LocalAddress;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.*;
import io.netty.handler.stream.ChunkedWriteHandler;
import io.netty.util.concurrent.DefaultPromise;
import io.netty.util.concurrent.FastThreadLocal;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Description:
 *
 *
 *  文件：
 *
 *      https(加密)、编码使用  chunk
 *
 *      默认使用 零拷贝 region
 *
 *
 *
 *  fastThreadLocal:
 *
 *      公用一个 EventExecutor 则是通过一个线程，http 请求又实现了重用该运行的 child channel eventLoop(实现了同一个线程的服用。)  + 独自的handler
 *
 *
 *
 *  channel handler:
 *      线程安全的，每一个对象对后方都会有自己的一个独有的handler。
 *
 *
 *
 *  1、一个客户端只能复用一个连接。
 *
 *          channel handler 线程安全的，每一个连接都是自己的一个独有的 handler 对象。
 *
 *          只有每一个客户端请求的handler，在active中初始化自己客户端的channel连接。才会复用线程。不然线程复用的是别人的线程。
 *
 *          ThreadLocal 在自己独有的handler中的才是线程安全的。
 *
 *          问题：
 *              存在多个channel。不能做到channel的复用。复用考虑在请求中增加请求上下文信息。
 *
 *
 * @author juntao.zhang
 * Date: 2018-07-19 下午8:59
 * @see
 */
@Slf4j
public class GatewayServer extends AbstractHttpServer{


    /** 公用一个 EventExecutor 则是通过一个线程，http 请求又实现了重用该运行的 child channel eventLoop(实现了同一个线程的服用。) **/
    public static final FastThreadLocal<DefaultPromise<String>> promise = new FastThreadLocal<>();


    public static void main(String[] args) {
        new GatewayServer().start(9878);
    }

    @Override
    public ChannelInitializer<NioSocketChannel> getChannelInitializer() {
        return new ChannelInitializer<NioSocketChannel>() {
            @Override
            protected void initChannel(NioSocketChannel ch) throws Exception {
                ChannelPipeline pipeline = ch.pipeline();
                pipeline.addLast(new HttpServerCodec());
                pipeline.addLast(new HttpContentCompressor());
                pipeline.addLast(new HttpObjectAggregator(1 << (10 * 2)));
                pipeline.addLast(new ChunkedWriteHandler());
//                pipeline.addLast(new MyHttpHandler());
                pipeline.addLast(new ForwardRequest());
            }
        };
    }



    public static final class MyHttpHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            super.userEventTriggered(ctx, evt);
        }

        @Override
        protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest msg) throws Exception {
            ctx.writeAndFlush(msg.retain());
        }
    }


    public static final class ForwardRequest extends SimpleChannelInboundHandler<FullHttpRequest>{

        private ChannelFuture channel;

        /**
         * 当连接被激活后，所属该连接的handler才会初始化下游连接。
         * @param ctx
         * @throws Exception
         */
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            channel = HttpClient.getChannel(ctx.channel().eventLoop(), new InetSocketAddress("localhost",8080));
            super.channelActive(ctx);
        }

        /**
         * 当该handler 对应的客户端关闭时候，才会关闭。
         * @param ctx
         * @throws Exception
         */
        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            channel.channel().close();
            super.channelInactive(ctx);
        }

        @Override
        protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest httpRequest) throws Exception {

            if (httpRequest.uri().contains("favicon.ico")){
                flushHttpResponse(ctx,httpRequest,"");
            }else{
                // 利用请求回调刷出请求
                DefaultPromise<String> objectDefaultPromise = new DefaultPromise<>(ctx.executor());
                // 当前线程的 threadlocal
                promise.set(objectDefaultPromise);
                objectDefaultPromise.addListener(future -> {
                    if (future.isSuccess()){
                        String content = (String) future.get();
                        flushHttpResponse(ctx,httpRequest,content);
                    }
                });


                log.info("open :{} threadid :{}",channel.channel().isOpen(),Thread.currentThread().getId());
                executHttpRequest(channel,"http://localhost:8080","shdfjdksfhs");
            }
        }
    }





    public static final class HttpClient{

        private static final ConcurrentHashMap<String,ChannelFuture> CHANNEL_POOL = new ConcurrentHashMap<>();


        public static ChannelFuture getChannel(EventLoop eventExecutors,InetSocketAddress socketAddress) throws InterruptedException {

            if (!CHANNEL_POOL.containsKey(generateKey(socketAddress))){

//                CHANNEL_POOL.put(generateKey(socketAddress),new Bootstrap()
//                        .option(ChannelOption.SO_KEEPALIVE, true)
//                        .option(ChannelOption.TCP_NODELAY, false)
//                        .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 4000)
//                        .channel(NioSocketChannel.class)
//                        .group(eventExecutors)
//                        .handler(new HttpClientChannelInitializer())
//                        .connect(socketAddress));

                return new Bootstrap()
                        .option(ChannelOption.SO_KEEPALIVE, true)
                        .option(ChannelOption.TCP_NODELAY, false)
                        .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 4000)
                        .channel(NioSocketChannel.class)
                        .group(eventExecutors)
                        .handler(new HttpClientChannelInitializer())
                        .connect(socketAddress);
            }
            ChannelFuture channelFuture = CHANNEL_POOL.get(generateKey(socketAddress));

            return reconnectChannelIfNeeds(channelFuture,eventExecutors,socketAddress);
        }

        public static String generateKey(InetSocketAddress socketAddress){
            return String.format("%s:%s",socketAddress.getHostName(),socketAddress.getPort());
        }


        /**
         * channel 保活，失败重新初始化
         * @param channelFuture
         * @param eventExecutors
         * @param socketAddress
         * @return
         * @throws InterruptedException
         */
        public static ChannelFuture reconnectChannelIfNeeds(ChannelFuture channelFuture,EventLoop eventExecutors,InetSocketAddress socketAddress) throws InterruptedException {
            if (Objects.nonNull(channelFuture)){
                Channel channel = channelFuture.channel();
                if (Objects.nonNull(channel)){
                    if (!channel.isOpen()){

                        CHANNEL_POOL.put(generateKey(socketAddress),
                                new Bootstrap()
                                        .option(ChannelOption.SO_KEEPALIVE, true)
                                        .option(ChannelOption.TCP_NODELAY, false)
                                        .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 4000)
                                        .channel(NioSocketChannel.class)
                                        .group(eventExecutors)
                                        .handler(new HttpClientChannelInitializer())
                                        .connect(socketAddress));

                        channel.close();
                        channelFuture =CHANNEL_POOL.get(generateKey(socketAddress));
                    }
                }else{
                    return channelFuture;
                }
            }
            return channelFuture;
        }


        public static final class HttpClientChannelInitializer extends ChannelInitializer<NioSocketChannel>{

            @Override
            protected void initChannel(NioSocketChannel ch) throws Exception {
                ChannelPipeline pipeline = ch.pipeline();
                pipeline.addLast(new HttpClientCodec());
                pipeline.addLast(new HttpObjectAggregator(1 << (10 * 2)));
                pipeline.addLast(new MyHttpClientHandler());
            }
        }


        public static final class MyHttpClientHandler extends SimpleChannelInboundHandler<Object> {

            @Override
            protected void channelRead0(ChannelHandlerContext ctx, Object object) throws Exception {


                log.info("thread id :{}",Thread.currentThread().getId());
                FullHttpResponse httpResponse = FullHttpResponse.class.cast(object);
                // DefaultPromise 输出配置
                DefaultPromise<String> stringDefaultPromise = promise.get();
                ByteBuf content = httpResponse.content();
                byte[] bytes = new byte[content.readableBytes()];
                content.readBytes(bytes);
                stringDefaultPromise.trySuccess(new String(bytes));
            }
        }
    }


    public static void flushHttpResponse(ChannelHandlerContext ctx, FullHttpRequest httpRequest,String msg) {
        // 写出响应
        boolean keepAlive = HttpUtil.isKeepAlive(httpRequest);
        DefaultFullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
        response.content().writeBytes(msg.getBytes());
        HttpHeaders headers = response.headers();
        headers.add("Content-Type","text/plain");

        // http body 字节数
        headers.addInt("content-length",response.content().readableBytes());

        // 告知请求方可以建立长连接
        headers.add("Connection","keep-alive");
        if (keepAlive){
            ctx.write(response);
        }else {
            // 实处响应后，并关闭连接
            ctx.write(response).addListener(ChannelFutureListener.CLOSE);
        }
        ctx.flush();
    }


    public static void executHttpRequest(ChannelFuture channelFuture, String url,String msg) {
        DefaultFullHttpRequest defaultFullHttpRequest =
                new DefaultFullHttpRequest(HttpVersion.HTTP_1_1,HttpMethod.GET,url, Unpooled.wrappedBuffer(msg.getBytes()));
        channelFuture.addListener((DefaultChannelPromise f) -> {
            if (f.isSuccess()){
                Channel channel = f.channel();
                channel.writeAndFlush(defaultFullHttpRequest);
            }else {
                log.error("执行http请求异常",f.cause());
            }
        });

    }
}
